package com.om.userpath.openeuler;

import com.om.Module.openeuler.logstash_repo_openeuler_org;
import com.om.common.CommonObject;
import com.om.userpath.utils.DataUtils;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructType;
import org.codehaus.jackson.JsonNode;
import org.elasticsearch.search.SearchHit;

import java.io.InputStream;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.*;

import static org.apache.spark.sql.functions.*;

public class UserPath {
    private static HashMap<String, String> platformMap = new HashMap();
    private static final String getIndexName;
    private static final String userPathHive;

    static {
        Properties properties = new Properties();
        platformMap = new DataUtils().getPlatformConfig(properties);
        getIndexName = properties.getProperty("openeuler.gitee.es.index");
        userPathHive = properties.getProperty("hive.openeuler.user.path");
    }

    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("TestHiveSql").enableHiveSupport().getOrCreate();
        registerUdf(spark);

        dataNginxLog(spark, args);
        dataGitee(spark, args);
        pushToEs(spark, args);
    }

    /**
     * gitee数据
     * 从ES拉取到Hive，并按时间做数据压缩
     *
     * @param spark sparkSession
     */
    private static void dataGitee(SparkSession spark, String[] args) {
        String startDate;
        String endDate;
        if (args == null || args.length < 2) {
            startDate = null;
            endDate = null;
        } else {
            startDate = args[0];
            endDate = args[1];
        }

        JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());
        List<SearchHit> hits = new DataUtils().searchAllFromEs(getIndexName, "_doc",
                new String[]{"created_at", "ip", "path", "author_name"}, new String[]{},
                "created_at", startDate, endDate, 60000L);
        System.out.println("************ " + hits.size() + " ************");
        ArrayList<Row> objects = new ArrayList<>();
        for (SearchHit hit : hits) {
            Map<String, Object> sourceAsMap = hit.getSourceAsMap();
            String t = (String) sourceAsMap.getOrDefault("created_at", null);
            String ip = (String) sourceAsMap.getOrDefault("ip", null);
            String path = (String) sourceAsMap.getOrDefault("path", null);
            String author_name = (String) sourceAsMap.getOrDefault("author_name", null);
            objects.add(RowFactory.create(t, ip, path, author_name));
        }

        JavaRDD<Row> rddRow = jsc.parallelize(objects);
        Dataset<Row> df = spark.createDataFrame(rddRow, StructType.fromDDL("created_at string,ip string,path string,author_name string"));

        Dataset<Row> dfr = df.filter("ip is not null")
                .withColumn("platform", lit("gitee"))
                .withColumn("created_at_minute", callUDF("minuteUDF", col("created_at")))
                .withColumn("created_at_hour", callUDF("hourUDF", col("created_at_minute")))
                .withColumn("created_at_day", callUDF("dayUDF", col("created_at_hour")))
                .withColumn("md5_key_minute", callUDF("keyUDF", col("ip"), col("platform"), col("created_at_minute")))
                .withColumn("md5_key_hour", callUDF("keyUDF", col("ip"), col("platform"), col("created_at_hour")))
                .withColumn("md5_key_day", callUDF("keyUDF", col("ip"), col("platform"), col("created_at_day")));

        dfr.write().mode("append").saveAsTable("openeuler." + userPathHive);
    }

    /**
     * 循环读取nginxlog文件
     *
     * @param spark sparkSession
     * @param args  时间段
     */
    private static void dataNginxLog(SparkSession spark, String[] args) {
        ArrayList<String> dates = getDates(args);
        for (String date : dates) {
            String file = "logstash-" + date + "-hk-app.json";
            System.out.println("****** start file:" + file + " ******");
            try {
                getLogData(spark, file);
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }
    }

    /**
     * 官网，repo下载，maillist数据
     * 直接从nginxLog中解析获取，并按时间做数据压缩，存入Hive
     *
     * @param spark sparkSession
     * @param file  文件名
     */
    private static void getLogData(SparkSession spark, String file) {
        String inputPath = "obs://om-collect/" + file;
        System.out.println("inputPath: " + inputPath);
        int taskCount = 100;
        String[] vhosts = new String[]{"repo.openeuler.org", "openeuler.org", "www.openeuler.org", "mailweb.openeuler.org"};

        JavaRDD<String> stringRDD = spark.sparkContext().textFile(inputPath, taskCount).toJavaRDD();

        JavaRDD javaRDD = (new logstash_repo_openeuler_org()).dealData(stringRDD, vhosts);

        JavaRDD<Row> javaRDD1 = javaRDD.map((Function<JsonNode, Row>) node -> {
            JsonNode sourceNode = node.get("_source");
            String logstr = sourceNode.get("log").getTextValue();
            JsonNode jsonData = CommonObject.objectMapper.readTree(logstr);

            String ip = (jsonData.get("proxy_remote_addr") == null) ? null : jsonData.get("proxy_remote_addr").getTextValue();
            if (StringUtils.isBlank(ip) || "-".equals(ip)) {
                ip = jsonData.get("remote_addr").getTextValue();
            }
            ip = (StringUtils.isBlank(ip) || "-".equals(ip)) ? null : ip;

            String t = jsonData.get("time").getTextValue();
            String path = jsonData.get("path").getTextValue();
            String platform = platformMap.get(jsonData.get("vhost").getTextValue());

            return RowFactory.create(t, ip, path, platform);
        });

        Dataset<Row> df = spark.createDataFrame(javaRDD1, StructType.fromDDL("created_at string,ip string,path string,platform string"));

        Dataset<Row> dfr = df.filter("ip is not null")
                .withColumn("author_name", lit(null).cast("string"))
                .withColumn("created_at_minute", callUDF("minuteUDF", col("created_at")))
                .withColumn("created_at_hour", callUDF("hourUDF", col("created_at_minute")))
                .withColumn("created_at_day", callUDF("dayUDF", col("created_at_hour")))
                .withColumn("md5_key_minute", callUDF("keyUDF", col("ip"), col("platform"), col("created_at_minute")))
                .withColumn("md5_key_hour", callUDF("keyUDF", col("ip"), col("platform"), col("created_at_hour")))
                .withColumn("md5_key_day", callUDF("keyUDF", col("ip"), col("platform"), col("created_at_day")));

        dfr.write().mode("append").saveAsTable("openeuler." + userPathHive);
    }

    /**
     * 注册自定义udf函数
     *
     * @param spark sparkSession
     */
    private static void registerUdf(SparkSession spark) {
        spark.sqlContext().udf().register("minuteUDF", (String s) -> {
            StringBuilder sb = new StringBuilder(s);
            return sb.replace(17, 19, "00").toString();
        }, DataTypes.StringType);

        spark.sqlContext().udf().register("hourUDF", (String s) -> {
            StringBuilder sb = new StringBuilder(s);
            return sb.replace(14, 16, "00").toString();
        }, DataTypes.StringType);

        spark.sqlContext().udf().register("dayUDF", (String s) -> {
            StringBuilder sb = new StringBuilder(s);
            return sb.replace(11, 13, "00").toString();
        }, DataTypes.StringType);

        spark.sqlContext().udf().register("keyUDF", (String s1, String s2, String s3) -> {
            return DigestUtils.md5Hex(s1 + s2 + s3);
        }, DataTypes.StringType);
    }

    /**
     * 获取日期范围内的所有日期
     *
     * @param args 时间段
     * @return 范围内所有日期
     */
    private static ArrayList<String> getDates(String[] args) {
        ArrayList<String> list = new ArrayList<>();
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy.MM.dd");

        LocalDate startDate;
        LocalDate endDate;
        if (args == null || args.length < 2) {
            startDate = LocalDate.now();
            endDate = LocalDate.now();
        } else {
            String startDateStr = args[0];
            String endDateStr = args[1];
            String[] sd = startDateStr.split("-");
            String[] ed = endDateStr.split("-");
            startDate = LocalDate.of(Integer.parseInt(sd[0]), Integer.parseInt(sd[1]), Integer.parseInt(sd[2]));
            endDate = LocalDate.of(Integer.parseInt(ed[0]), Integer.parseInt(ed[1]), Integer.parseInt(ed[2]));
        }

        LocalDate dateBegin = startDate;
        while (dateBegin.compareTo(endDate) <= 0) {
            list.add(dateBegin.format(formatter));
            dateBegin = dateBegin.plusDays(1);
        }

        return list;
    }

    /**
     * 将hive数据推推送到ES
     *
     * @param spark sparkSession
     * @param args  时间段
     */
    private static void pushToEs(SparkSession spark, String[] args) {
        String sqlStr;
        if (args == null || args.length < 2) {
            sqlStr = "select * from openeuler." + userPathHive;
        } else {
            String startTime = args[0] + "T00:00:00+00:00";
            String endTime = args[1] + "T23:59:59+00:00";
            sqlStr = String.format("select * from openeuler." + userPathHive +
                    " where created_at >= timestamp('%s') and created_at <= timestamp('%s')", startTime, endTime);
        }

        Dataset<Row> df = spark.sql(sqlStr)
                .select("ip", "author_name", "created_at_minute", "platform", "md5_key_minute")
                .dropDuplicates("md5_key_minute");

        new DataUtils().pushToEs(df, "md5_key_minute", "openeuler_user_path_20201120", "_doc");
    }
}
